package org.example.mq;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.example.exception.InitConsumerException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @author: wf
 * @create: 2020/12/8 15:42
 * @description: 消费者
 */
public class ConsumerProxy {
    private static final Logger log = LoggerFactory.getLogger(ConsumerProxy.class);
    private static final int DEFAULT_RETRY = 5;
    private static final int CORE_THREAD_SIZE = 5;
    private static final int MAX_THREAD_SIZE = 20;
    private static final int PULL_THRESHOLD_QUEUE = 50;
    
    private Map<String, ConsumerServer> topics;
    private String namesrv;
    private DefaultMQPushConsumer consumer;
    private String groupName;
    
    /**
     * 构造器(默认集群模式)
     * @param topics
     * @param namesrv
     * @param groupName
     */
    public ConsumerProxy(Map<String, ConsumerServer> topics, String namesrv, String groupName) {
        ConsumerServer consumerServer = topics.values().stream().filter(f -> StringUtils.isEmpty(f.getTopic())).findFirst().orElse(null);
        if (Objects.nonNull(consumerServer)){
            throw new InitConsumerException("consumerServer topic is null：" + consumerServer.getClass().getName());
        }
        this.topics = topics;
        this.namesrv = namesrv;
        this.groupName = groupName;
        initConsumer(MAX_THREAD_SIZE, DEFAULT_RETRY, null, 0L, CORE_THREAD_SIZE);
    }
    
    /**
     * 构造器(默认集群模式)
     * @param topicList
     * @param namesrv
     * @param groupName
     */
    public ConsumerProxy(List<ConsumerServer> topicList, String namesrv, String groupName) {
        ConsumerServer consumerServer = topicList.stream().filter(f -> StringUtils.isEmpty(f.getTopic())).findFirst().orElse(null);
        if (Objects.nonNull(consumerServer)){
            throw new InitConsumerException("consumerServer topic is null：" + consumerServer.getClass().getName());
        }
        this.topics = topicList.stream().collect(Collectors.toMap(ConsumerServer::getTopic, f -> f));
        this.namesrv = namesrv;
        this.groupName = groupName;
        initConsumer(MAX_THREAD_SIZE, DEFAULT_RETRY, null, 0L, CORE_THREAD_SIZE);
    }

    /**
     * 构造器(指定集群模式)
     * @param topicList
     * @param messageModel
     * @param namesrv
     * @param groupName
     */
    public ConsumerProxy(List<ConsumerServer> topicList, MessageModel messageModel, String namesrv, String groupName) {
        ConsumerServer consumerServer = topicList.stream().filter(f -> StringUtils.isEmpty(f.getTopic())).findFirst().orElse(null);
        if (Objects.nonNull(consumerServer)){
            throw new InitConsumerException("consumerServer topic is null：" + consumerServer.getClass().getName());
        }
        this.topics = topicList.stream().collect(Collectors.toMap(ConsumerServer::getTopic, f -> f));
        this.namesrv = namesrv;
        this.groupName = groupName;
        initConsumer(MAX_THREAD_SIZE, DEFAULT_RETRY, messageModel, 0L, CORE_THREAD_SIZE);
    }

    /**
     * 消费者初始化
     * @param maxThreadSize
     * @param retryTime
     * @param messageModel
     * @param consumeTimeout
     * @param coreThreadSize
     */
    private void initConsumer(int maxThreadSize, int retryTime, MessageModel messageModel, long consumeTimeout, int coreThreadSize) {
        if (this.topics == null) {
            throw new InitConsumerException("topics is null");
        }
        if (StringUtils.isEmpty(this.namesrv)) {
            throw new InitConsumerException("namesrv is null");
        }
        initConsumerCommon(maxThreadSize, retryTime, messageModel, consumeTimeout, coreThreadSize);
        try {
            Set<Map.Entry<String, ConsumerServer>> entries = this.topics.entrySet();
            for (Map.Entry<String, ConsumerServer> entry : entries) {
                ConsumerServer server = entry.getValue();
                String topic = server.getTopic();
                String tags = StringUtils.isEmpty(server.getTagExp()) ? "*" : server.getTagExp();
                this.consumer.subscribe(topic, tags);
            }
            this.consumer.registerMessageListener(this.genMessageListener());
            this.consumer.start();
        } catch (Exception e) {
            log.error("init the consumer error", e);
            throw new InitConsumerException(e.getMessage());
        }
    }

    /**
     * 初始化消费者参数
     *
     * @param maxThreadSize
     * @param retryTime
     * @param messageModel
     * @param consumeTimeout
     * @param coreThreadSize
     */
    private void initConsumerCommon(int maxThreadSize, int retryTime, MessageModel messageModel, long consumeTimeout, int coreThreadSize) {
        this.consumer = new DefaultMQPushConsumer(this.groupName);
        this.consumer.setNamesrvAddr(this.namesrv);
        this.consumer.setConsumeThreadMax(maxThreadSize);
        this.consumer.setConsumeThreadMin(coreThreadSize);
        this.consumer.setMaxReconsumeTimes(retryTime);
        this.consumer.setPullThresholdForQueue(PULL_THRESHOLD_QUEUE);
        if (messageModel != null) {
            this.consumer.setMessageModel(messageModel);
        }
        if (consumeTimeout > 0L) {
            this.consumer.setConsumeTimeout(consumeTimeout);
        }
        this.consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
    }

    /**
     * topic监听器
     * @return
     */
    private MessageListenerConcurrently genMessageListener() {
        return (msgs, context) -> {
            for (int i = 0; i < msgs.size(); i++) {
                ConsumerServer consumerServer = this.topics.get(msgs.get(i).getTopic());
                if (consumerServer == null){
                    log.error("no topic = {} for service", msgs.get(i).getTopic());
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                try {
                    if (!consumerServer.receive(msgs.get(i)))
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                } catch (Exception e) {
                    log.error("consumer message error", e);
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        };
    }
}
